package com.shanshan.wheel.rpc.common.remoting.transport.codec;

/**
 * 自定义协议
 * 消息头 16B
 *
 * 4B  magic code（魔法数） 1B version（版本）   4B full length（消息长度[头长 + 消息体长]） 1B messageType（消息类型）
 * 1B codec（序列化类型） 1B compress（压缩类型）  4B  requestId（请求的Id）
 * body（object类型数据）
 */

import com.shanshan.wheel.rpc.common.compress.GzipCompress;
import com.shanshan.wheel.rpc.common.remoting.constant.RpcConstants;
import com.shanshan.wheel.rpc.common.remoting.constant.RpcMessageType;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcMessage;
import com.shanshan.wheel.rpc.common.serialize.KryoSerializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Description 编码器
 * @ClassName RpcMessageEncoder
 * @Author 山山大魔王
 * @Date 2022/9/4 17:50
 * @Version 1.0
 */
public class RpcMessageEncoder extends MessageToByteEncoder<RpcMessage> {

    private final AtomicInteger REQUEST_ID = new AtomicInteger(0);

    @Override
    protected void encode(ChannelHandlerContext ctx, RpcMessage msg, ByteBuf out) throws Exception {
        // 设置自定义协议魔术
        out.writeBytes(RpcConstants.MAGIC_NUMBER);
        // 设置自定义协议版本
        out.writeByte(RpcConstants.VERSION);
        // 预留长度字节
        out.writerIndex(out.writerIndex() + 4);
        byte messageType = msg.getMessageType();
        // 设置消息类型
        out.writeByte(messageType);
        // 设置序列化类型
        out.writeByte(msg.getCodec());
        // 设置压缩类型
        out.writeByte(msg.getCompress());
        // 设置请求Id
        out.writeInt(REQUEST_ID.getAndIncrement());

        byte[] bodyBytes = null;

        int fullLength = RpcConstants.HEAD_LENGTH;

        if( messageType != RpcMessageType.HEARTBEAT_PING_TYPE.getCode() &&
            messageType != RpcMessageType.HEARTBEAT_PONG_TYPE.getCode() ){
            Object data = msg.getData();
            // 序列化
            KryoSerializer kryoSerializer = new KryoSerializer();
            byte[] serialize = kryoSerializer.serialize(data);
            // 数据压缩
            GzipCompress gzipCompress = new GzipCompress();
            bodyBytes = gzipCompress.compress(serialize);
            fullLength += bodyBytes.length;
        }

        if(bodyBytes != null){
            out.writeBytes(bodyBytes);
        }
        // 获取当前写索引
        int writeIndex = out.writerIndex();
        // 设置写索引 到协议版本后面的位置
        out.writerIndex(writeIndex - fullLength + RpcConstants.MAGIC_NUMBER.length + 1);
        // 写入消息整个帧的长度
        out.writeInt(fullLength);
        // 将写索引设置为原来的位置
        out.writerIndex(writeIndex);
        int i = out.readableBytes();
    }
}
